package com.wenbing.miio.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;

/**
 * 基于ServerSocketChannel实现的NetworkInterface
 * @author wen bing
 * @projectName: miio
 * @package: com.wenbing.miio.core
 * @className: DataAdapter
 * @date 2024/09/21 12:41
 **/
public class ServerSocketChannelNetworkInterface implements NetworkInterface {
    private static final Logger LOGGER = LoggerFactory.getLogger(ServerSocketChannelNetworkInterface.class);

    public static final String BIND_CHANNEL = "bind_socket_channel";

    private Selector selector;

    private ServerSocketChannel channel;

    private Thread readThread;

    private BufferQueue<RowFrame> frameBufferQueue;

    public ServerSocketChannelNetworkInterface(int port) throws IOException {
        this(new InetSocketAddress("0.0.0.0", port));
    }

    public ServerSocketChannelNetworkInterface(InetSocketAddress address) throws IOException {
        this.channel = ServerSocketChannel.open();
        this.channel.bind(address);
        this.selector = Selector.open();
        this.channel.configureBlocking(false);
        this.channel.register(selector, SelectionKey.OP_ACCEPT);

        this.frameBufferQueue = new BufferQueue<>();
    }

    private void run() throws IOException {
        selector.select();
        Set<SelectionKey> keys = selector.selectedKeys();
        for (SelectionKey key : keys) {
            if (key.isAcceptable()) {
                ServerSocketChannel channel = (ServerSocketChannel)key.channel();
                SocketChannel socketChannel = channel.accept();
                socketChannel.configureBlocking(false);
                socketChannel.register(selector, SelectionKey.OP_READ);
                continue;
            }
            if (key.isReadable()) {
                SocketChannel channel = (SocketChannel)key.channel();
                readFromChannel(channel);
            }
        }
    }

    @Override
    public void boot() {
        this.readThread = InnerThread.daemonThread("NetworkInterface-serverSocketChannel", this::run);
        this.readThread.start();
        LOGGER.info("{} start success ", this.readThread.getName());
        LOGGER.info("ServerSocketChannelNetworkInterface boot success bind address is {}", getAddress());
    }
    private void readFromChannel(SocketChannel channel) throws IOException {
        NetworkLayerContext context = MapNetworkLayerContext.rootContext(channel.getLocalAddress(), channel.getRemoteAddress());
        context.setShareAttribute(BIND_CHANNEL, channel);
        ByteBuffer bf = ByteBuffer.allocate(512);
        channel.read(bf);
        frameBufferQueue.unshift(new RowFrame(context, bf));
    }

    @Override
    public SocketAddress getAddress() {
        try {
            return this.channel.getLocalAddress();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public RowFrame receiveRowFrame() throws IOException {
        try {
            return frameBufferQueue.shift();
        } catch (InterruptedException interruptedException) {
            throw new IOException(interruptedException);
        }
    }

    @Override
    public void sendRowFrame(RowFrame rowFrame) throws IOException {
        ByteBuffer bf = rowFrame.getByteBuffer();
        SocketChannel channel = rowFrame.getContext().getShareAttribute(BIND_CHANNEL);
        if (channel.isOpen() && channel.isConnected()) {
            channel.write(bf);
        }
    }
}
